package com.buddy.zookeeper.worker;

import com.buddy.zookeeper.master.MasterAsync;
import org.apache.zookeeper.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.List;

public class WorkreWatcher implements Watcher {

    private static final Logger LOG = LoggerFactory.getLogger(MasterAsync.class);


    ZooKeeper zk;
    String hostPort;
    ChildrenCache workersCache;
    AsyncCallback.ChildrenCallback workersGetChildrenCallback = new AsyncCallback.ChildrenCallback() {
        public void processResult(int rc, String path, Object ctx, List<String> children) {
            switch (KeeperException.Code.get(rc)){
                case CONNECTIONLOSS:
                    getWorkers();
                    break;
                case OK:
                    LOG.info("Successfully got a list of worders:"+
                    children.size()+
                    " workers");
                    reassignAndSet(children);
                    break;
                default:
                    LOG.error("getChildren failed",
                            KeeperException.create(KeeperException.Code.get(rc),path));
            }
        }
    };


    Watcher workersChangeWatcher = new Watcher() {
        public void process(WatchedEvent event) {
            if (event.getType() == Event.EventType.NodeChildrenChanged){
                assert "/workers".equals(event.getPath());
                getWorkers();
            }
        }
    };

    WorkreWatcher(String hostPort) {
        this.hostPort = hostPort;
    }


    void getWorkers(){
        zk.getChildren("/workers",
                workersChangeWatcher,
                workersGetChildrenCallback,null);
    }


    void startZK() throws IOException {
        zk = new ZooKeeper(hostPort, 15000, this);
    }
    void stopZK() throws InterruptedException {
        zk.close();
    }

    void reassignAndSet(List<String> children) {
        List<String> toProcess;
        if (workersCache == null){
            workersCache = new ChildrenCache(children);
            toProcess = null;
        }else{
            LOG.info("Removing and setting");
            toProcess = workersCache.removeAndSet(children);
        }
        if (toProcess != null){
            for (String worker:toProcess){
                getAbsentWorkerTasks(worker);
            }
        }
    }

    void getAbsentWorkerTasks(String worker) {
    }

    public void process(WatchedEvent event) {

    }
}
